package com.zbmy.plugin.im.server.core.back;

import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import java.util.Vector;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.log4j.Logger;

import com.base.framework.utils.JsonUtils;
import com.zbmy.plugin.im.server.core.protocol.MessageHolder;
import com.zbmy.plugin.im.server.core.queue.TaskQueue;

/**
 * 
 * 业务服务层  
 * @ClassName: BackLogicService    
 * @author 杨乐平    
 * @date 2018年4月28日 下午6:04:10    
 * @version  v 1.0
 */
public class BackLogicService {
	Logger logger = Logger.getLogger(BackLogicService.class);
	public static AtomicBoolean shutdown = new AtomicBoolean(false);
    // 任务队列
    private BlockingQueue<MessageHolder> taskQueue;
    // 阻塞式地从taskQueue取MessageHolder
    private ExecutorService takeExecutor;
    // 执行业务的线程池
    private ExecutorService taskExecutor;
    // 消息业务监听
    private Vector<MessageListener> messageListenerVector = new Vector<MessageListener>();//监听自己的监听器队列     

    public void addDemoListener(MessageListener dl) {     
    	messageListenerVector.addElement(dl);     
    }     
	public void notifyDemoEvent(MessageEvent messageEvent) {//通知所有的监听器     
	        Enumeration<MessageListener> enum1 = messageListenerVector.elements(); 
	        while(enum1.hasMoreElements()) {     
	        	MessageListener dl = (MessageListener)enum1.nextElement();     
	     	   dl.handleEvent(messageEvent);
	        }     
	}

    public void startUp() {
    	messageListenerVector.add(new CustomMessageListener());
    	takeExecutor = Executors.newSingleThreadExecutor();
        taskExecutor = Executors.newFixedThreadPool(20);
        taskQueue = TaskQueue.getQueue();
        takeExecutor.execute(new Runnable() {
            @Override
            public void run() {
                while (!shutdown.get()) {
                    try {
                        MessageHolder messageHolder = taskQueue.take();
                        startTask(messageHolder);
                    } catch (InterruptedException e) {
                        logger.warn("receiveQueue take", e);
                    }
                }
            }

            private void startTask(MessageHolder messageHolder) {
                taskExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                    	try {
                    		Map<String, Object> extraMsgMap = new HashMap<String, Object>();
                        	MessageEvent messageEvent  = new MessageEvent(this, messageHolder, extraMsgMap);
                        	notifyDemoEvent(messageEvent);
						} catch (Exception e) {
							logger.error("发布消息事件出现异常，messageHolder:{}"+JsonUtils.writeValueAsString(messageHolder), e);
						}
                    }
                });
            }
        });
    }
}
